package com.atguigu.flink.processFunction;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Process {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);

        SingleOutputStreamOperator<WaterSensor> mapDS = socketDS.map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor ws, long l) {
                                        return ws.getTs();
                                    }
                                }));

        //按照传感器的id进行分组
        KeyedStream<WaterSensor, String> keyByDS = mapDS.keyBy(WaterSensor::getId);

        keyByDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor waterSensor, KeyedProcessFunction<String, WaterSensor, String>.Context context, Collector<String> collector) throws Exception {
                        String currentKey = context.getCurrentKey();
                        System.out.println("currentKey = " + currentKey);
                        Long timestamp = context.timestamp();
                        System.out.println("timestamp = " + timestamp);
                        //将当前数据放入侧输出流
                        //context.output();
                        TimerService timerService = context.timerService();
                        long currentProcessingTime = timerService.currentProcessingTime();
                        System.out.println("currentProcessingTime = " + currentProcessingTime);
                        //当前水位线
                        long currentWatermark = timerService.currentWatermark();
                        System.out.println("currentWatermark = " + currentWatermark);
                        //注册事件时间定时器
                        timerService.registerEventTimeTimer(10);
                    }

                }
        );
    }
}
